package cn.iocoder.yudao.module.system.tcpserver.protocol1;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.HexUtil;
import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.module.system.dal.dataobject.zkdata.ZkDataDO;
import cn.iocoder.yudao.module.system.tcpserver.pojo.ZkPacketFrame;
import io.netty.channel.*;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.ScheduledFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;


/**
 * 说明：处理器
 */
@ChannelHandler.Sharable
@Component
@Slf4j
public class ProtocolServerHandler extends SimpleChannelInboundHandler<Object> {

    @Resource
    private ZkDataProcessService zkDataService;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object obj)
            throws Exception {
        Channel incoming = ctx.channel();

        // 如果不是protobuf类型的数据
        if (!(obj instanceof ZkDataDO)) {
            log.info("未知数据!" + obj);
            return;
        }
        ZkDataDO msg = (ZkDataDO) obj;
        try {

            log.info("Client->Server:{}-- oriData:{}", msg.getDataSender(), msg.getOriData());
            log.info("Client->Server:{}--msgJson:{}", msg.getDataSender(), JSONUtil.toJsonStr(msg));

            ZkPacketFrame packetFrame = zkDataService.process(msg);
            if (packetFrame != null && packetFrame.packet() != null) {
                ctx.writeAndFlush(packetFrame);
                ScheduledFuture<?> future = ctx.executor().scheduleAtFixedRate(
                        () -> zkDataService.sendServerHeartbeat(ctx.channel(), msg.getDataSender()),
                        5, 5, TimeUnit.SECONDS // 5秒后首次发送，之后每5秒一次
                );

                if (HeartbeatServer.getClientMap().containsKey(msg.getDataSender())) {
                    HeartbeatServer.ChannelSchedule schedule = HeartbeatServer.getClientMap().remove(msg.getDataSender());
                    if (schedule != null && schedule.future != null) {
                        schedule.future.cancel(true);
                        log.info("111心跳任务关闭：{}", msg.getDataSender());
                    }
                    if (schedule != null && schedule.channel != null) {
                        schedule.channel.close();
                        log.info("111连接关闭：{}", msg.getDataSender());
                    }
                }

                HeartbeatServer.getClientMap().put(msg.getDataSender(),
                        new HeartbeatServer.ChannelSchedule(ctx.channel(), future)
                );
            }
            //todo send data
            List<ZkDataDO> dataToSend = zkDataService.getDataToSend(msg.getDataSender());
            if (CollUtil.isNotEmpty(dataToSend)) {
                dataToSend.forEach(data -> {
                    ctx.writeAndFlush(new ZkPacketFrame(HexUtil.decodeHex(data.getOriData())));
                    zkDataService.updateStatus(data);
                });
            }

//            else {
//                log.error("数据解析失败,plaload:{}", msg.getOriData());
//                ChannelFuture channelFuture = ctx.writeAndFlush("ERROR DATA".getBytes()).addListener(ChannelFutureListener.CLOSE);
//                channelFuture.addListener(new ChannelFutureListener() {
//                    @Override
//                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
//                        log.info("errordata-channelFuture closed:{}", channelFuture.channel());
//                    }
//                });
//            }

        } catch (Exception e) {
            log.error("channelRead0 error {}", obj);
            log.error("channelRead0 error", e);
        } finally {
            ReferenceCountUtil.release(obj);
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        log.info("连接断开");
        // 客户端断开时清理资源
//        HeartbeatServer.getClientMap().values().removeIf(s -> s.channel == ctx.channel());
        for (String s : HeartbeatServer.getClientMap().keySet()) {
            if (HeartbeatServer.getClientMap().get(s) != null &&
                    HeartbeatServer.getClientMap().get(s).channel == ctx.channel()) {
                HeartbeatServer.ChannelSchedule schedule = HeartbeatServer.getClientMap().remove(s);
                if (schedule != null && schedule.future != null) {
                    schedule.future.cancel(true);
                }
                zkDataService.clientOffLine(s);
                log.info("连接断开：{}", s);
            }
        }
        ctx.fireChannelInactive();
    }


}
